package org.example.common;

import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Schema;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Random;
import java.util.concurrent.TimeUnit;

/**
 * @author zhangjg
 * @desc Producer
 * @date 2021/12/22 22:00
 **/
public class ProducerP {

    private final static Logger log = LoggerFactory.getLogger(ProducerP.class);

    public static void main(String[] args) throws Exception{

        PulsarClient client = PulsarClient.builder()
                .serviceUrl("pulsar://localhost:6650")
                .build();
        Producer<String> stringProducer = client.newProducer(Schema.STRING)
                .topic("persistent://public/default/tt11")
                .producerName("p1")
                .sendTimeout(10,TimeUnit.SECONDS)
//                .compressionType(CompressionType.LZ4)
//                .batchingMaxPublishDelay(1,TimeUnit.SECONDS)
                .create();
        String strLine = null;
        Random random = new Random();
        int k = 0;
        for (int i=1;i<100;i++) {
            k = random.nextInt(1000);
            strLine = "My message i="+i+" time="+ 8;
            stringProducer.newMessage(Schema.STRING)
//                    .deliverAfter(60,TimeUnit.SECONDS)
//                    .value("My message i="+i+" time="+ DateUtil.now())
                    .value(strLine)
//                    .sequenceId(k)
                    .sendAsync();
            log.info(" msg = {},sequenceId={}",strLine,i);
            Thread.sleep(TimeUnit.SECONDS.toMillis(3));
        }


        Thread.sleep(TimeUnit.SECONDS.toMillis(300));
        stringProducer.sendAsync("xx");
        stringProducer.closeAsync()
                .thenRun(() -> {
                    System.out.println("Producer closed");
                    try {
                        client.close();
                    } catch (PulsarClientException e) {
                        e.printStackTrace();
                    }
                })
                .exceptionally((ex) -> {
                    System.err.println("Failed to close producer: " + ex);
                    return null;
                });
    }
}
